// Copyright(C) 2024 InfiniFlow, Inc. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

module;

#include "base64.hpp"

module infinity_core:meta_key.impl;

import :meta_key;
import :utility;
import :kv_code;
import :infinity_exception;
import :meta_type;
import :default_values;

import std;
import third_party;

import column_def;

namespace infinity {

ColumnMetaKey::ColumnMetaKey(std::string db_id_str,
                             std::string table_id_str,
                             SegmentID segment_id,
                             BlockID block_id,
                             std::shared_ptr<ColumnDef> column_def)
    : MetaKey(MetaType::kBlockColumn), db_id_str_(std::move(db_id_str)), table_id_str_(std::move(table_id_str)), segment_id_(segment_id),
      block_id_(block_id), column_def_(std::move(column_def)) {}

ColumnMetaKey::~ColumnMetaKey() = default;

std::string DBMetaKey::ToString() const { return fmt::format("db: {}:{}", KeyEncode::CatalogDbKey(db_name_, commit_ts_), db_id_str_); }

std::string DBTagMetaKey::ToString() const { return fmt::format("db_tag: {}:{}", KeyEncode::CatalogDbTagKey(db_id_str_, tag_name_), value_); }

std::string TableMetaKey::ToString() const {
    return fmt::format("table: {}:{}", KeyEncode::CatalogTableKey(db_id_str_, table_name_, commit_ts_), table_id_str_);
}

std::string TableNameMetaKey::ToString() const {
    return fmt::format("table name: {}:{}", KeyEncode::CatalogTableKey(db_id_str_, table_name_, commit_ts_), table_id_str_);
}

std::string TableColumnMetaKey::ToString() const {
    return fmt::format("table_column: {}:{}", KeyEncode::TableColumnKey(db_id_str_, table_id_str_, column_name_, commit_ts_), value_);
}
std::string TableTagMetaKey::ToString() const {
    return fmt::format("table_tag: {}:{}", KeyEncode::CatalogTableTagKey(db_id_str_, table_id_str_, tag_name_), value_);
}

std::string SegmentMetaKey::ToString() const {
    return fmt::format("segment: {}:{}", KeyEncode::CatalogTableSegmentKey(db_id_str_, table_id_str_, segment_id_), commit_ts_);
}

std::string SegmentTagMetaKey::ToString() const {
    if (tag_name_ == "fast_rough_filter") {
        return fmt::format("segment_tag: {}", KeyEncode::CatalogTableSegmentTagKey(db_id_str_, table_id_str_, segment_id_, tag_name_));
    }
    return fmt::format("segment_tag: {}:{}", KeyEncode::CatalogTableSegmentTagKey(db_id_str_, table_id_str_, segment_id_, tag_name_), value_);
}

std::string BlockMetaKey::ToString() const {
    return fmt::format("block: {}:{}", KeyEncode::CatalogTableSegmentBlockKey(db_id_str_, table_id_str_, segment_id_, block_id_), commit_ts_);
}

std::string BlockTagMetaKey::ToString() const {
    if (tag_name_ == "fast_rough_filter") {
        return fmt::format("block_tag: {}", KeyEncode::CatalogTableSegmentBlockTagKey(db_id_str_, table_id_str_, segment_id_, block_id_, tag_name_));
    }
    return fmt::format("block_tag: {}:{}",
                       KeyEncode::CatalogTableSegmentBlockTagKey(db_id_str_, table_id_str_, segment_id_, block_id_, tag_name_),
                       value_);
}

std::string ColumnMetaKey::ToString() const { return fmt::format("block column: not implemented"); }

std::string TableIndexMetaKey::ToString() const {
    return fmt::format("table_index: {}:{}", KeyEncode::CatalogIndexKey(db_id_str_, table_id_str_, index_name_, commit_ts_), index_id_str_);
}

std::string TableIndexTagMetaKey::ToString() const {
    return fmt::format("table_index_tag: {}:{}", KeyEncode::CatalogIndexTagKey(db_id_str_, table_id_str_, index_id_str_, tag_name_), value_);
}

std::string SegmentIndexMetaKey::ToString() const {
    return fmt::format("segment_index: {}:{}", KeyEncode::CatalogIdxSegmentKey(db_id_str_, table_id_str_, index_id_str_, segment_id_), commit_ts_);
}

std::string SegmentIndexTagMetaKey::ToString() const {
    return fmt::format("segment_index_tag: {}:{}",
                       KeyEncode::CatalogIdxSegmentTagKey(db_id_str_, table_id_str_, index_id_str_, segment_id_, tag_name_),
                       value_);
}

std::string ChunkIndexMetaKey::ToString() const {
    return fmt::format("chunk_index: {}:{}",
                       KeyEncode::CatalogIdxChunkKey(db_id_str_, table_id_str_, index_id_str_, segment_id_, chunk_id_),
                       commit_ts_);
}

std::string ChunkIndexTagMetaKey::ToString() const {
    return fmt::format("chunk_index_tag: {}:{}",
                       KeyEncode::CatalogIdxChunkTagKey(db_id_str_, table_id_str_, index_id_str_, segment_id_, chunk_id_, tag_name_),
                       value_);
}

std::string SystemTagMetaKey::ToString() const { return fmt::format("system_tag: {}:{}", tag_name_, value_); }

std::string PmObjectMetaKey::ToString() const { return fmt::format("pm_path: {}:{}", KeyEncode::PMObjectKey(path_key_), value_); }

std::string PmStatMetaKey::ToString() const { return fmt::format("pm_object: {}:{}", KeyEncode::PMObjectStatKey(object_key_), value_); }

std::string DropMetaKey::ToString() const { return fmt::format("drop_key: drop|{}|{}:{}", scope_, object_key_, value_); }

nlohmann::json DBMetaKey::ToJson() const {
    nlohmann::json json_res;
    json_res["db_id"] = std::stoull(db_id_str_);
    json_res["db_name"] = db_name_;
    if (commit_ts_ != UNCOMMIT_TS) {
        json_res["commit_ts"] = commit_ts_;
    }

    return json_res;
}

nlohmann::json DBTagMetaKey::ToJson() const {
    nlohmann::json json_res;
    json_res[tag_name_] = value_;
    return json_res;
}

nlohmann::json TableMetaKey::ToJson() const {
    nlohmann::json json_res;
    json_res["table_id"] = std::stoull(table_id_str_);
    json_res["table_name"] = table_name_;
    if (commit_ts_ != UNCOMMIT_TS) {
        json_res["commit_ts"] = commit_ts_;
    }
    return json_res;
}

nlohmann::json TableNameMetaKey::ToJson() const {
    nlohmann::json json_res;
    json_res["table_id"] = std::stoull(table_id_str_);
    json_res["table_name"] = table_name_;
    if (commit_ts_ != UNCOMMIT_TS) {
        json_res["commit_ts"] = commit_ts_;
    }
    return json_res;
}

nlohmann::json TableColumnMetaKey::ToJson() const {
    nlohmann::json json_res;
    json_res["column_definition"] = nlohmann::json::parse(value_);
    return json_res;
}

nlohmann::json TableTagMetaKey::ToJson() const {
    nlohmann::json json_res;
    json_res[tag_name_] = nlohmann::json::parse(value_);
    return json_res;
}

nlohmann::json SegmentMetaKey::ToJson() const {
    nlohmann::json json_res;
    json_res["segment_id"] = segment_id_;
    if (commit_ts_ != UNCOMMIT_TS) {
        json_res["commit_ts"] = commit_ts_;
    }
    return json_res;
}

nlohmann::json SegmentTagMetaKey::ToJson() const {
    nlohmann::json json_res;
    json_res["tag_name"] = tag_name_;
    // TODO: fast rough filter isn't deserialized here.
    //    json_res["tag_value"] = value_;
    return json_res;
}

nlohmann::json BlockMetaKey::ToJson() const {
    nlohmann::json json_res;
    json_res["block_id"] = block_id_;
    if (commit_ts_ != UNCOMMIT_TS) {
        json_res["commit_ts"] = commit_ts_;
    }
    return json_res;
}

nlohmann::json BlockTagMetaKey::ToJson() const {
    nlohmann::json json_res;
    json_res[tag_name_] = base64::to_base64(value_);
    return json_res;
}

nlohmann::json ColumnMetaKey::ToJson() const {
    nlohmann::json json_res;
    return json_res;
}

nlohmann::json TableIndexMetaKey::ToJson() const {
    nlohmann::json json_res;
    json_res["index_id"] = index_id_str_;
    json_res["index_name"] = index_name_;
    if (commit_ts_ != UNCOMMIT_TS) {
        json_res["commit_ts"] = commit_ts_;
    }
    return json_res;
}

nlohmann::json TableIndexTagMetaKey::ToJson() const {
    nlohmann::json json_res;

    // Handle empty or invalid JSON gracefully
    if (value_.empty()) {
        // If value is empty, set it as an empty JSON object
        json_res[tag_name_] = nlohmann::json::object();
    } else {
        try {
            // Try to parse the JSON, if it fails, use the raw string value
            json_res[tag_name_] = nlohmann::json::parse(value_);
        } catch (const nlohmann::json::parse_error &) {
            // If parsing fails, use the raw string value
            json_res[tag_name_] = value_;
        }
    }

    return json_res;
}

nlohmann::json SegmentIndexMetaKey::ToJson() const {
    nlohmann::json json_res;
    json_res["segment_id"] = segment_id_;
    if (commit_ts_ != UNCOMMIT_TS) {
        json_res["commit_ts"] = commit_ts_;
    }
    return json_res;
}

nlohmann::json SegmentIndexTagMetaKey::ToJson() const {
    nlohmann::json json_res;
    json_res[tag_name_] = nlohmann::json::parse(value_);
    return json_res;
}

nlohmann::json ChunkIndexMetaKey::ToJson() const {
    nlohmann::json json_res;
    json_res["chunk_id"] = chunk_id_;
    if (commit_ts_ != UNCOMMIT_TS) {
        json_res["commit_ts"] = commit_ts_;
    }
    return json_res;
}

nlohmann::json ChunkIndexTagMetaKey::ToJson() const {
    nlohmann::json json_res;
    json_res[tag_name_] = nlohmann::json::parse(value_);
    return json_res;
}

nlohmann::json SystemTagMetaKey::ToJson() const {
    nlohmann::json json_res;
    json_res[tag_name_] = nlohmann::json::parse(value_);
    return json_res;
}

nlohmann::json PmObjectMetaKey::ToJson() const {
    nlohmann::json json_res;
    json_res["path"] = path_key_;
    json_res["description"] = nlohmann::json::parse(value_);
    return json_res;
}

nlohmann::json PmStatMetaKey::ToJson() const {
    nlohmann::json json_res;
    json_res["object"] = object_key_;
    json_res["stat"] = nlohmann::json::parse(value_);
    return json_res;
}

nlohmann::json DropMetaKey::ToJson() const {
    nlohmann::json json_res;
    json_res["scope"] = scope_;
    json_res["key"] = object_key_;
    json_res["value"] = nlohmann::json::parse(value_);
    return json_res;
}

std::shared_ptr<MetaKey> MetaParse(const std::string &key, const std::string &value) {
    std::vector<std::string> fields = infinity::Partition(key, '|');

    auto fields_size = fields.size();

    if (fields[0] == "catalog" && fields[1] == "db") {
        const std::string &db_name_str = fields[2];
        const std::string &commit_ts_str = fields[3];
        const std::string &db_id_str = value;
        auto db_meta_key = std::make_shared<DBMetaKey>(db_id_str, db_name_str);
        db_meta_key->commit_ts_ = std::stoull(commit_ts_str);
        return db_meta_key;
    }

    if (fields[0] == "catalog" && fields[1] == "tbl") {
        const std::string &db_id_str = fields[2];
        const std::string &table_name_str = fields[3];
        const std::string &commit_ts_str = fields[4];
        const std::string &table_id_str = value;

        auto table_meta_key = std::make_shared<TableMetaKey>(db_id_str, table_id_str, table_name_str);
        table_meta_key->commit_ts_ = std::stoull(commit_ts_str);
        return table_meta_key;
    }

    if (fields[0] == "catalog" && fields[1] == "seg") {
        const std::string &db_id_str = fields[2];
        const std::string &table_id_str = fields[3];
        const std::string &segment_id_str = fields[4];
        const std::string &commit_ts_str = value;
        SegmentID segment_id = std::stoul(segment_id_str);
        auto segment_meta_key = std::make_shared<SegmentMetaKey>(db_id_str, table_id_str, segment_id);
        segment_meta_key->commit_ts_ = std::stoull(commit_ts_str);
        return segment_meta_key;
    }

    // construct segment tag meta key
    if (fields[0] == "seg") {
        const std::string &db_id_str = fields[1];
        const std::string &table_id_str = fields[2];
        const std::string &segment_id_str = fields[3];
        const std::string &tag_name_str = fields[4];
        SegmentID segment_id = std::stoul(segment_id_str);
        auto segment_tag_meta_key = std::make_shared<SegmentTagMetaKey>(db_id_str, table_id_str, segment_id, tag_name_str);
        segment_tag_meta_key->value_ = value;
        return segment_tag_meta_key;
    }

    // construct blk tag meta key
    if (fields[0] == "blk") {
        const std::string &db_id_str = fields[1];
        const std::string &table_id_str = fields[2];
        const std::string &segment_id_str = fields[3];
        const std::string &block_id_str = fields[4];
        const std::string &tag_name_str = fields[5];
        SegmentID segment_id = std::stoul(segment_id_str);
        BlockID block_id = std::stoul(block_id_str);
        auto block_tag_meta_key = std::make_shared<BlockTagMetaKey>(db_id_str, table_id_str, segment_id, block_id, tag_name_str);
        block_tag_meta_key->value_ = value;
        return block_tag_meta_key;
    }

    if (fields[0] == "catalog" && fields[1] == "blk") {
        const std::string &db_id_str = fields[2];
        const std::string &table_id_str = fields[3];
        const std::string &segment_id_str = fields[4];
        const std::string &block_id_str = fields[5];
        const std::string &commit_ts_str = value;
        SegmentID segment_id = std::stoul(segment_id_str);
        BlockID block_id = std::stoul(block_id_str);
        auto block_meta_key = std::make_shared<BlockMetaKey>(db_id_str, table_id_str, segment_id, block_id);
        block_meta_key->commit_ts_ = std::stoull(commit_ts_str);
        return block_meta_key;
    }

    if (fields[0] == "db") {
        const std::string &db_id_str = fields[1];
        const std::string &tag_name_str = fields[2];
        auto db_tag_meta_key = std::make_shared<DBTagMetaKey>(db_id_str, tag_name_str);
        db_tag_meta_key->value_ = value;
        return db_tag_meta_key;
    }

    if (fields[0] == "tbl") {
        if (fields[1] == "col") {
            const std::string &db_id_str = fields[2];
            const std::string &table_id_str = fields[3];
            const std::string &column_name_str = fields[4];
            const std::string &commit_ts_str = fields[5];
            auto table_column_meta_key = std::make_shared<TableColumnMetaKey>(db_id_str, table_id_str, column_name_str);
            table_column_meta_key->commit_ts_ = std::stoull(commit_ts_str);
            table_column_meta_key->value_ = value;
            return table_column_meta_key;
        }
        const std::string &db_id_str = fields[1];
        const std::string &table_id_str = fields[2];
        const std::string &tag_name_str = fields[3];
        auto table_tag_meta_key = std::make_shared<TableTagMetaKey>(db_id_str, table_id_str, tag_name_str);
        table_tag_meta_key->value_ = value;
        return table_tag_meta_key;
    }

    if (fields[0] == "catalog" && fields[1] == "idx") {
        const std::string &db_id_str = fields[2];
        const std::string &table_id_str = fields[3];
        const std::string &index_name_str = fields[4];
        const std::string &commit_ts_str = fields[5];
        const std::string &index_id_str = value;

        auto table_index_meta_key = std::make_shared<TableIndexMetaKey>(db_id_str, table_id_str, index_id_str, index_name_str);
        table_index_meta_key->commit_ts_ = std::stoull(commit_ts_str);
        return table_index_meta_key;
    }

    if (fields[0] == "idx") {
        const std::string &db_id_str = fields[1];
        const std::string &table_id_str = fields[2];
        const std::string &index_id_str = fields[3];
        const std::string &tag_name_str = fields[4];
        auto table_index_tag_meta_key = std::make_shared<TableIndexTagMetaKey>(db_id_str, table_id_str, index_id_str, tag_name_str);
        table_index_tag_meta_key->value_ = value;
        return table_index_tag_meta_key;
    }

    if (fields[0] == "idx_seg") {
        const std::string &db_id_str = fields[1];
        const std::string &table_id_str = fields[2];
        const std::string &index_id_str = fields[3];
        SegmentID segment_id = std::stoull(fields[4]);
        if (fields_size == 6) {
            const std::string &tag_name_str = fields[5];
            auto segment_index_tag_meta_key =
                std::make_shared<SegmentIndexTagMetaKey>(db_id_str, table_id_str, index_id_str, segment_id, tag_name_str);
            segment_index_tag_meta_key->value_ = value;
            return segment_index_tag_meta_key;
        }
        auto segment_index_meta_key = std::make_shared<SegmentIndexMetaKey>(db_id_str, table_id_str, index_id_str, segment_id);
        segment_index_meta_key->commit_ts_ = std::stoull(value);
        return segment_index_meta_key;
    }

    if (fields[0] == "idx_chunk") {
        const std::string &db_id_str = fields[1];
        const std::string &table_id_str = fields[2];
        const std::string &index_id_str = fields[3];
        SegmentID segment_id = std::stoull(fields[4]);
        ChunkID chunk_id = std::stoull(fields[5]);
        if (fields_size == 7) {
            const std::string &tag_name_str = fields[6];
            auto chunk_index_tag_meta_key =
                std::make_shared<ChunkIndexTagMetaKey>(db_id_str, table_id_str, index_id_str, segment_id, chunk_id, tag_name_str);
            chunk_index_tag_meta_key->value_ = value;
            return chunk_index_tag_meta_key;
        }
        auto chunk_index_meta_key = std::make_shared<ChunkIndexMetaKey>(db_id_str, table_id_str, index_id_str, segment_id, chunk_id);
        chunk_index_meta_key->commit_ts_ = std::stoull(value);
        return chunk_index_meta_key;
    }

    if (fields[0] == "pm") {
        if (fields[1] == "object") {
            const std::string &path_key = fields[2];
            std::shared_ptr<PmObjectMetaKey> pm_path_meta_key = std::make_shared<PmObjectMetaKey>(path_key);
            pm_path_meta_key->value_ = value; //
            return pm_path_meta_key;
        }
        if (fields[1] == "object_stat") {
            const std::string &object_key = fields[2];
            auto pm_object_meta_key = std::make_shared<PmStatMetaKey>(object_key);
            pm_object_meta_key->value_ = value;
            return pm_object_meta_key;
        }
        UnrecoverableError(fmt::format("Unexpected key: {}:{}", key, value));
    }

    if (fields[0] == "drop") {
        const std::string &scope = fields[1];
        const std::string &object_key = fields[2];
        auto drop_meta_key = std::make_shared<DropMetaKey>(scope, object_key);
        drop_meta_key->value_ = value;
        return drop_meta_key;
    }

    const std::string &tag_name_str = fields[0];
    auto system_tag_meta_key = std::make_shared<SystemTagMetaKey>(tag_name_str);
    system_tag_meta_key->value_ = value;
    return system_tag_meta_key;
}

} // namespace infinity